﻿
Imports std
Imports storoomc
Imports storcore
Imports storcore.storoomDistrTM

Public Class nodes_redundance2
    Inherits nodes_redundance

    Public Shared ReadOnly defaultTimeStamp As Int64 = minInt64
    'to save memory, just let the background syncer read the data again
    Private syncqueue As setqueue(Of String) = Nothing
    Private syncerTimer3Index As Guid = Guid.Empty

    Private MustInherit Class tsdata
        Implements IComparable(Of tsdata), IComparable
        Public ts As Int64 = defaultTimeStamp
        Public MustOverride Function size() As Int64

        Public Function CompareTo(ByVal obj As Object) As Integer Implements IComparable.CompareTo
            Return CompareTo(convertor.cast(Of tsdata)(obj))
        End Function

        'though compare can handle me is nothing / other is nothing, and some other cases, but i think this way is safer
        Public Function CompareTo(ByVal other As tsdata) As Integer Implements IComparable(Of tsdata).CompareTo
            'as the defaultTimeStamp is minInt64 and hasnoTimeStamp is minInt32,
            'and all the other possible ts is possitive, so just compare the int64 is enough
            If Me Is Nothing AndAlso other Is Nothing Then
                Return 0
            ElseIf Me Is Nothing Then
                Return -1
            ElseIf other Is Nothing Then
                Return 1
            Else
                If ts < other.ts Then
                    Return -1
                ElseIf ts > other.ts Then
                    Return 1
                Else
                    If size() < other.size() Then
                        Return -1
                    ElseIf size() > other.size() Then
                        Return 1
                    Else
                        Return 0
                    End If
                End If
            End If
        End Function
    End Class

    'do not make tsdata1 ICloneable, so we can save a pointer here
    Private Class tsdata1
        Inherits tsdata
        Implements IComparable(Of tsdata), IComparable, IComparable(Of tsdata1), IComparable(Of tsdata2)
        Public buff As bytes = Nothing

        Public Sub New()
        End Sub

        Public Sub New(ByVal ts As Int64, ByVal buff As bytes)
            Me.ts = ts
            Me.buff = buff
        End Sub

        Public Sub New(ByVal buff As bytes, ByVal ts As Int64)
            Me.New(ts, buff)
        End Sub

        Public NotOverridable Overrides Function size() As Int64
            If buff Is Nothing Then
                Return 0
            Else
                Return buff.size()
            End If
        End Function

        Public Overloads Function CompareTo(ByVal other As tsdata1) As Integer _
                                                                    Implements IComparable(Of tsdata1).CompareTo
            Return MyBase.CompareTo(convertor.cast(Of tsdata)(other))
        End Function

        Public Overloads Function CompareTo(ByVal other As tsdata2) As Integer _
                                                                    Implements IComparable(Of tsdata2).CompareTo
            Return MyBase.CompareTo(convertor.cast(Of tsdata)(other))
        End Function
    End Class

    'a faked data with only buffsize, instead of buff
    Private Class tsdata2
        Inherits tsdata
        Implements IComparable, IComparable(Of tsdata), IComparable(Of tsdata1), IComparable(Of tsdata2)
        Public _size As Int64 = 0

        Public Sub New()
        End Sub

        Public Sub New(ByVal size As Int64, ByVal ts As Int64)
            Me._size = size
            Me.ts = ts
        End Sub

        Public NotOverridable Overrides Function size() As Int64
            Return _size
        End Function

        Public Overloads Function CompareTo(ByVal other As tsdata1) As Integer _
                                                                    Implements IComparable(Of tsdata1).CompareTo
            Return MyBase.CompareTo(convertor.cast(Of tsdata)(other))
        End Function

        Public Overloads Function CompareTo(ByVal other As tsdata2) As Integer _
                                                                    Implements IComparable(Of tsdata2).CompareTo
            Return MyBase.CompareTo(convertor.cast(Of tsdata)(other))
        End Function
    End Class

    'syncerStore needs thread protected
    'cannot confirm whether timer2/timer3 always make call in ONE thread.
    Private syncerStore As tsdata1 = Nothing

    Protected NotOverridable Overrides Function slicehealth(ByVal slice As storageInterface) As Boolean
        'do nothing when slicehealth, though do clear slicehealthHandler.
        Return True
    End Function

    Private Delegate Function wmDelegate(ByVal key As String, ByVal buff As bytes) As Boolean

    Private Function _storageInterfaceWMImpl(ByVal si As storageInterface, ByVal key As String, _
                                             ByVal buff As bytes, ByVal ts As Int64, _
                                             ByVal wm As wmDelegate) As Boolean
        assert(Not si Is Nothing, "storageInterface is nothing.")
        assert(Not wm Is Nothing, "wmDelegate is nothing.")
        Return wm(datakey(key), buff) AndAlso si.modify(timestampkey(key), convertor.tobytes(ts))
    End Function

    Protected Overridable Function _storageInterfaceModifyImpl(ByVal si As storageInterface, ByVal key As String, _
                                                               ByVal buff As bytes, ByVal ts As Int64) As Boolean
        assert(Not si Is Nothing)
        Return _storageInterfaceWMImpl(si, key, buff, ts, AddressOf si.modify)
    End Function

    Protected Overridable Function _storageInterfaceWriteImpl(ByVal si As storageInterface, ByVal key As String, _
                                                              ByVal buff As bytes, ByVal ts As Int64) As Boolean
        assert(Not si Is Nothing)
        Return _storageInterfaceWMImpl(si, key, buff, ts, AddressOf si.write)
    End Function

    Private Delegate Function wmDelegate2(ByVal si As storageInterface, ByVal key As String, _
                                          ByVal buff As bytes, ByVal ts As Int64) As Boolean

    Private Function _storageInterfaceWMImpl(ByVal si As storageInterface, ByVal key As String, _
                                             ByVal d As tsdata1, ByVal wm As wmDelegate2) As Boolean
        assert(Not d Is Nothing, "d is nothing.")
        assert(Not wm Is Nothing)
        Return wm(si, key, d.buff, d.ts)
    End Function

    Private Function _storageInterfaceModifyImpl(ByVal si As storageInterface, ByVal key As String, _
                                                 ByVal d As tsdata1) As Boolean
        Return _storageInterfaceWMImpl(si, key, d, AddressOf _storageInterfaceModifyImpl)
    End Function

    Private Function _storageInterfaceWriteImpl(ByVal si As storageInterface, ByVal key As String, _
                                                ByVal d As tsdata1) As Boolean
        Return _storageInterfaceWMImpl(si, key, d, AddressOf _storageInterfaceWriteImpl)
    End Function

    Private Function storageInterfaceReadTimeStampImpl(ByVal si As storageInterface, ByVal key As String, _
                                                       ByRef ts As Int64) As Boolean
        assert(Not si Is Nothing, "storageInterface is nothing.")
        Dim tsbuff As bytes = Nothing
        If Not si.read(timestampkey(key), tsbuff) OrElse Not convertor.toint64(tsbuff, ts) Then
            Return False
        Else
            Return True
        End If
    End Function

    Protected Overridable Function _storageInterfaceReadImpl(ByVal si As storageInterface, ByVal key As String, _
                                                             ByRef buff As bytes, ByRef ts As Int64) As Boolean
        assert(Not si Is Nothing, "storageInterface is nothing.")
        If Not storageInterfaceReadTimeStampImpl(si, key, ts) Then
            'let ts be the oldest timestamp, so each other data can overwrite it
            ts = defaultTimeStamp
        End If
        If Not si.read(datakey(key), buff) Then
            ts = defaultTimeStamp
            Return False
        Else
            Return True
        End If
    End Function

    Private Function storageInterfaceReadImpl(ByVal si As storageInterface, ByVal key As String, _
                                              ByRef d As tsdata1) As Boolean
        If d Is Nothing Then
            d = New tsdata1()
        End If

        Return _storageInterfaceReadImpl(si, key, d.buff, d.ts)
    End Function

    Private Class syncerReadInfo
        Inherits distrInfo(Of String, Boolean)
        Private key As String = Nothing
        Private d As tsdata1 = Nothing

        'this is single-thread mode, so only need to confirm not syncerStore is nothing.
        Public NotOverridable Overrides Function finishOperation(ByRef rtn As Boolean) As Boolean
            If result = ternary.true AndAlso Not d Is Nothing Then
                Dim store As tsdata1 = Nothing
                store = convertor.cast(Of nodes_redundance2)(caller).syncerStore
                If compare(store, d) < 0 Then
                    store.buff = d.buff
                    store.ts = d.ts
                End If

                rtn = True
            Else
                'only modify when all the slices returns normal
                rtn = False
            End If

            Return True
        End Function

        Protected NotOverridable Overrides Function initialInfo(ByVal index As Integer, ByVal arg As String) As Boolean
            If MyBase.initialInfo(index, arg) Then
                copy(key, arg)
                Return True
            Else
                Return False
            End If
        End Function

        Protected NotOverridable Overrides Function operationImpl() As ternary
            Return ifoper(convertor.cast(Of nodes_redundance2)(caller).storageInterfaceReadImpl(slice, key, d), _
                          ternary.true, ternary.false)
        End Function

        Public Shared Function createInfo() As syncerReadInfo
            Return New syncerReadInfo()
        End Function
    End Class

    Private Class syncerModifyInfo
        Inherits wmInfo

        Public NotOverridable Overrides Function finishOperation(ByRef rtn As Boolean) As Boolean
            rtn = rtn AndAlso result = ternary.true
            Return True
        End Function

        Protected NotOverridable Overrides Function operationImpl() As ternary
            assert(Not slice Is Nothing, "slice is nothing.")
            If v Is Nothing Then
                Return ternary.false
            Else

            End If
            Dim thists As Int64 = defaultTimeStamp
            Dim size As UInt64 = 0
            If Not convertor.cast(Of nodes_redundance2)(caller). _
                       storageInterfaceReadTimeStampImpl(slice, key, thists) OrElse _
               Not slice.sizeof(datakey(key), size) OrElse _
               compare(New tsdata2(size, thists), v) < 0 Then
                If Not convertor.cast(Of nodes_redundance2)(caller). _
                           _storageInterfaceModifyImpl(slice, key, v) Then
                    raiseError("cannot modify slice in syncerModify, slice " + _
                               convertor.cast(Of nodes_redundance2)(caller).showslice(slice) + ", key " + key, _
                               errorHandle.errorType.warning)
                    Return ternary.false
                Else
                    Return ternary.true
                End If
            ElseIf compare(New tsdata2(size, thists), v) > 0 Then
                raiseError("got a timestamp or size larger than what syncerRead got, " + _
                           "assume has been changed already. key " + key, errorHandle.errorType.warning)
                Return ternary.unknown
            Else
                Return ternary.true
            End If
        End Function

        Public Shared Shadows Function createInfo() As syncerModifyInfo
            Return New syncerModifyInfo()
        End Function
    End Class

    Private Sub insertSyncQueue(ByVal key As String)
        SyncLock syncqueue
            syncqueue.push(key)
        End SyncLock
    End Sub

    Private Function clearSyncQueue() As Boolean
        SyncLock syncqueue
            syncqueue.clear()
        End SyncLock
        Return True
    End Function

    Private Sub syncer()
        assert(Not syncerStore Is Nothing, "syncerStore has not been initialized before syncer call starts.")
        Dim k As String = Nothing
        If Not syncqueue.empty() Then
            SyncLock syncqueue
                If Not syncqueue.empty() Then
                    k = syncqueue.front()
                    syncqueue.pop()
                End If
            End SyncLock

            If Not isemptystring(k) Then
                If isdebugmode() Then
                    raiseError("sync key " + k)
                End If
                Dim rtn As Boolean = False
                'cannot confirm whether timer3 makes two calls with different threads at a certain time
                SyncLock syncerStore
                    syncerStore.buff = Nothing
                    syncerStore.ts = defaultTimeStamp
                    If Not tm.distrOperation(Of syncerReadInfo, String, Boolean) _
                                            (k, rtn, AddressOf syncerReadInfo.createInfo, slices) Then
                        raiseError("cannot finish distrOperation when syncerRead.", errorHandle.errorType.exclamation)
                        If seekImpl(k, slices) = ternary.true Then
                            insertSyncQueue(k)
                        End If
                        'do not try again if cannot read & seek, the data may be not available any more,
                        'such as deleted already.
                        rtn = False
                    ElseIf rtn Then
                        If Not wm_redundancy(wmInfo.wmtype.modify, k, syncerStore, slices, _
                                             AddressOf syncerModifyInfo.createInfo, rtn) OrElse Not rtn Then
                            raiseError("cannot syncerModify for key " + k, errorHandle.errorType.warning)
                            'let's try again later if cannot modify
                            insertSyncQueue(k)
                        End If
                    Else
                        raiseError("cannot syncerRead for key " + k, errorHandle.errorType.warning)
                        'do not try again if cannot read & seek, the data may be not available any more,
                        'such as deleted already.
                        If seekImpl(k, slices) = ternary.true Then
                            insertSyncQueue(k)
                        End If
                    End If
                End SyncLock
            End If
        End If
    End Sub

    Private Class wmInfo
        Inherits wmInfoBase(Of tsdata1)

        Protected Overrides Function operationImpl() As ternary
            If v Is Nothing Then
                Return ternary.false
            Else
                Dim rst As Boolean
                If wm = wmInfoBase(Of tsdata1).wmtype.write Then
                    rst = convertor.cast(Of nodes_redundance2)(caller). _
                              _storageInterfaceWriteImpl(slice, key, v)
                Else
                    rst = convertor.cast(Of nodes_redundance2)(caller). _
                              _storageInterfaceModifyImpl(slice, key, v)
                End If
                Return ifoper(rst, ternary.true, ternary.false)
            End If
        End Function

        Public Shared Function createInfo() As wmInfo
            Return New wmInfo()
        End Function
    End Class

    Protected NotOverridable Overrides Function write_redundancy(ByVal key As String, ByVal buff As bytes, _
                                                                 ByVal slices As vector(Of storageInterface)) As Boolean
        Return wm_redundancy(wmInfo.wmtype.write, key, _
                             New tsdata1(buff, Now.Ticks()), slices, AddressOf wmInfo.createInfo)
    End Function

    Protected NotOverridable Overrides Function modify_redundancy(ByVal key As String, ByVal buff As bytes, _
                                                                  ByVal slices As vector(Of storageInterface)) _
                                                                 As Boolean
        Return wm_redundancy(wmInfo.wmtype.modify, key, _
                             New tsdata1(buff, Now.Ticks()), slices, AddressOf wmInfo.createInfo)
    End Function

    Private Shadows Class readInfo
        Inherits distrInfo(Of String, tsdata1)
        Protected key As String = Nothing
        Protected d As tsdata1 = Nothing

        Private Sub insertSyncQueue()
            convertor.cast(Of nodes_redundance2)(caller).insertSyncQueue(key)
        End Sub

        Public NotOverridable Overrides Function finishOperation(ByRef rtn As tsdata1) As Boolean
            'logic
            '1. if cannot read from slice,
            '   - if rtn.ts is defaultTimeStamp, 
            '     means there is no other slice finished before, set rtn.ts to hasnoTimeStamp
            '   - if rtn.ts is not hasnoTimeStamp,
            '     means there is another slice finished before, and set the rtn.ts to something else, insertSyncQueue
            '2. if can read from slice,
            '   - if rtn.ts is not ts,
            '       - if rtn.ts is not defaultTimeStamp, or rtn.second is hasnoTimeStamp, 
            '         means some other slices have set rtn.ts, insertSyncQueue
            '   - if rtn.ts < ts or rtn.ts is defaultTimeStamp or rtn.ts is hasnoTimeStamp,
            '         update rtn with buff and ts
            'if result is true, ts should not be defaultTimeStamp, but make it safer
            If result = ternary.true AndAlso Not d Is Nothing AndAlso d.ts <> defaultTimeStamp Then
                'no other before
                If rtn Is Nothing Then
                    rtn = New tsdata1(d.ts, d.buff)
                Else
                    Dim cmp As Int32 = 0
                    cmp = compare(rtn, d)
                    If cmp <> 0 Then
                        insertSyncQueue()
                        If cmp < 0 Then
                            rtn.ts = d.ts
                            rtn.buff = d.buff
                        End If
                    End If
                End If
            Else
                If rtn Is Nothing Then
                    rtn = New tsdata1(defaultTimeStamp, Nothing)
                ElseIf rtn.ts <> defaultTimeStamp Then
                    insertSyncQueue()
                End If
            End If

            Return True
        End Function

        Protected NotOverridable Overrides Function initialInfo(ByVal index As Integer, ByVal arg As String) As Boolean
            If MyBase.initialInfo(index, arg) Then
                copy(key, arg)
                Return True
            Else
                Return False
            End If
        End Function

        Protected NotOverridable Overrides Function operationImpl() As ternary
            Return ifoper(convertor.cast(Of nodes_redundance2)(caller).storageInterfaceReadImpl(slice, key, d), _
                          ternary.true, ternary.false)
        End Function

        Public Shared Function createInfo() As readInfo
            Return New readInfo()
        End Function
    End Class

    Protected NotOverridable Overrides Function readImpl(ByVal key As String, ByRef buff As bytes, _
                                                         ByVal slices As vector(Of storageInterface)) As Boolean
        Dim d As tsdata1 = Nothing
        If Not oper.distrOperation(Of readInfo, String, tsdata1) _
                                  (key, d, AddressOf readInfo.createInfo, slices) Then
            raiseError("cannot finish distrOperation when read", errorHandle.errorType.exclamation)
            Return False
        ElseIf d Is Nothing OrElse d.buff Is Nothing Then
            Return False
        Else
            buff = d.buff
            Return True
        End If
    End Function

    Protected Overrides Function seekwork(ByVal slice As storageInterface, _
                                          ByRef key As String) As Boolean
        Return MyBase.seekwork(slice, datakey(key)) AndAlso MyBase.seekwork(slice, timestampkey(key))
    End Function

    Protected Overrides Function listwork(ByVal slice As storageInterface, _
                                          ByRef arg As vector(Of String)) As Boolean
        Dim v As vector(Of String) = Nothing
        If MyBase.listwork(slice, v) Then
            If arg Is Nothing Then
                arg = New vector(Of String)()
            Else
                arg.clear()
            End If
            Dim i As Int64
            For i = 0 To v.size() - 1
                If isdatakey(v(i)) Then
                    arg.push_back(originalkey(v(i)))
                End If
            Next

            Return True
        Else
            Return False
        End If
    End Function

    Protected Overrides Function deleteImpl(ByVal key As String, _
                                            ByVal slices As vector(Of storageInterface)) As Boolean
        Return MyBase.deleteImpl(datakey(key), slices) AndAlso MyBase.deleteImpl(timestampkey(key), slices)
    End Function

    Protected Class sizeofInfo
        Inherits storoomDistrTM.sizeofInfo

        Private Sub insertSyncQueue()
            convertor.cast(Of nodes_redundance2)(caller).insertSyncQueue(key)
        End Sub

        Public NotOverridable Overrides Function finishOperation(ByRef rtn As pair(Of UInt64, Boolean)) As Boolean
            'logic
            'if this info is success
            '   if no other rtn before
            '       just set size of this info to rtn, and mark as success
            '   else if rtn.first <> size orelse rtn.second marked as failed
            '       insert sync queue
            'else if this info was failed
            '   if no other rtn before
            '       just set size of rtn to 0, and mark as failed
            '   else if rtn.second marked as success, means there is at least one slice returned success with size
            '       insert sync queue
            If result = ternary.true Then
                If rtn Is Nothing Then
                    rtn = make_pair(Of UInt64, Boolean)(size, True)
                Else
                    If rtn.first <> size OrElse Not (rtn.second) Then
                        insertSyncQueue()
                        rtn.second = True
                        If rtn.first < size Then
                            rtn.first = size
                        End If
                    End If
                End If
            Else
                If rtn Is Nothing Then
                    rtn = make_pair(Of UInt64, Boolean)(0, False)
                ElseIf rtn.second Then
                    insertSyncQueue()
                End If
            End If

            Return True
        End Function

        Public Shared Shadows Function createInfo() As sizeofInfo
            Return New sizeofInfo()
        End Function

        Protected Overrides Function operationImpl() As ternary
            Return ifoper(slice.sizeof(datakey(key), size), ternary.true, ternary.false)
        End Function
    End Class

    Protected Overrides Function sizeofImpl(ByVal key As String, ByRef size As UInt64, _
                                            ByVal slices As vector(Of storageInterface)) As Boolean
        Return sizeofImpl(key, size, slices, AddressOf sizeofInfo.createInfo)
    End Function

    Public Overloads Sub startImpl()
        MyBase.startImpl(False)
        syncerTimer3Index = timer3.addcall(AddressOf syncer, secondsToMilliseconds(1))
        sliceHealthHandle = Nothing
    End Sub

    Public NotOverridable Overrides Sub start()
        nodesStart(Me)
        startImpl()
    End Sub

    Protected NotOverridable Overrides Function closeImpl(ByVal slices As vector(Of storageInterface)) As Boolean
        If syncerTimer3Index <> Guid.Empty Then
            debugassert(timer3.removecall(syncerTimer3Index), "cannot remove timer3 call")
            syncerTimer3Index = Guid.Empty
        End If
        Return MyBase.closeImpl(slices)
    End Function

    Protected Overrides Function keycountwork(ByVal slice As storageInterface, ByRef count As UInt64) As Boolean
        If MyBase.keycountwork(slice, count) Then
            count >>= 1
            Return True
        Else
            Return False
        End If
    End Function

    Protected Overrides Function retireImpl(ByVal slices As vector(Of storageInterface)) As Boolean
        Return MyBase.retireImpl(slices) AndAlso clearSyncQueue()
    End Function

    Public Sub New()
        syncqueue = New setqueue(Of String)()
        syncerStore = New tsdata1()
    End Sub
End Class
